package net.fruit.mq.client.impl;

import java.util.Properties;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionCheckListener;
import org.apache.rocketmq.common.message.MessageExt;

import net.fruit.mq.client.Constants;
import net.fruit.mq.client.Consumer;
import net.fruit.mq.client.MQFactoryAPI;
import net.fruit.mq.client.Message;
import net.fruit.mq.client.Producer;
import net.fruit.mq.client.impl.rocketmq.ConsumerImpl;
import net.fruit.mq.client.impl.rocketmq.OrderConsumerImpl;
import net.fruit.mq.client.impl.rocketmq.OrderProducerImpl;
import net.fruit.mq.client.impl.rocketmq.ProducerImpl;
import net.fruit.mq.client.impl.rocketmq.RMQUtil;
import net.fruit.mq.client.impl.rocketmq.TransactionProducerImpl;
import net.fruit.mq.client.order.OrderConsumer;
import net.fruit.mq.client.order.OrderProducer;
import net.fruit.mq.client.transaction.LocalTransactionChecker;
import net.fruit.mq.client.transaction.TransactionProducer;
import net.fruit.mq.client.transaction.TransactionStatus;

public class MQFactoryImpl implements MQFactoryAPI {

    @Override
    public Producer createProducer(final Properties properties) {
        ProducerImpl producer = new ProducerImpl(properties);
        return producer;
    }


    @Override
    public Consumer createConsumer(final Properties properties) {
        ConsumerImpl consumer = new ConsumerImpl(properties);
        return consumer;
    }


    @Override
    public OrderProducer createOrderProducer(final Properties properties) {
        OrderProducerImpl producer = new OrderProducerImpl(properties);
        return producer;
    }


    @Override
    public OrderConsumer createOrderedConsumer(final Properties properties) {
        OrderConsumerImpl consumer = new OrderConsumerImpl(properties);
        return consumer;
    }


    @Override
    public TransactionProducer createTransactionProducer(Properties properties,
                                                         final LocalTransactionChecker checker) {
        TransactionProducerImpl transactionProducer = new TransactionProducerImpl(properties, new TransactionCheckListener() {
            @Override
            public LocalTransactionState checkLocalTransactionState(MessageExt msg) {
                String msgId = msg.getProperty(Constants.TRANSACTION_ID);
                Message message = RMQUtil.msgConvert(msg);
                message.setMsgID(msgId);
                TransactionStatus check = checker.check(message);
                if (TransactionStatus.CommitTransaction == check) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                } else if (TransactionStatus.RollbackTransaction == check) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                return LocalTransactionState.UNKNOW;
            }
        });
        return transactionProducer;
    }
}
